package cn.monkey.io.transport.netty.duplex;

import cn.monkey.io.transport.core.Outbound;
import cn.monkey.io.transport.netty.NettyInbound;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;

import java.util.List;

public class SimpleDuplexInbound implements DuplexInbound {
    protected final NettyInbound nettyInbound;
    protected final Outbound duplexOutbound;
    protected final Scheduler scheduler;
    protected final Disposable disposable;

    public SimpleDuplexInbound(NettyInbound nettyInbound,
                               Outbound duplexOutbound,
                               Scheduler scheduler) {
        this.nettyInbound = nettyInbound;
        this.duplexOutbound = duplexOutbound;
        this.scheduler = scheduler;
        this.disposable = this.init();
    }

    protected Disposable init() {
        return this.nettyInbound()
                .receive(Object.class)
                .subscribeOn(this.scheduler)
                .subscribe(this.outbound()::writeAndFlush);
    }


    @Override
    public NettyInbound nettyInbound() {
        return this.nettyInbound;
    }

    @Override
    public Outbound outbound() {
        return this.duplexOutbound;
    }

    @Override
    public Mono<Void> close() {
        this.disposable.dispose();
        List<Mono<Void>> list = List.of(this.nettyInbound().close(), this.outbound().close());
        return Flux.merge(list).then();
    }
}
